package com.shujia.mr.tuijian;

import com.shujia.mr.wc1.WordCountDemo1;
import com.shujia.mr.wc1.WordCountMapper;
import com.shujia.mr.wc1.WordCountReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/*
    输入：
        tom hello hadoop cat
        hadoop hello cat

    输出：
        <"tom-hello", 1>
        <"tom-hadoop", 1>
        <"tom-cat", 1>
        <"hello-hadoop", 0>
        <"hello-cat", 0>
        <"hadoop-cat", 0>
        ....
        <"tom-hello", 0>
        <"hadoop-hello", 1>


    进入reduce逻辑：
        <"tom-hello",[1,0,1,0,1,1]>
        <"hello-hadoop",[0,1]>


 */
public class PersonDemo {
    public static void main(String[] args) {
        try {
            //创建hadoop配置文件对象
            Configuration conf = new Configuration();

            //创建MapReduce作业对象
            Job job = Job.getInstance(conf);
            //设置mr作业的名字
            job.setJobName("33期好友推荐案例1");
            //设置reduce的个数
//            job.setNumReduceTasks(2);

            //设置mr作业的主类
            job.setJarByClass(PersonDemo.class);

            //设置map类
            job.setMapperClass(PersonMapper.class);
            //设置reduce类
            job.setReducerClass(PersonReducer.class);

            //设置map任务的输出数据类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);

            //设置reduce任务的输出类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            // 设置读取hdfs文件的路径
//            FileInputFormat.addInputPath(); //添加某一个目录
//            FileInputFormat.setInputPaths(); // 可以添加多个目录
//            FileInputFormat.setInputPaths(job, new Path("/xxx/xxx/xx"));
            FileInputFormat.setInputPaths(job, new Path(args[0]));

            //设置输出的hdfs目录路径
            FileOutputFormat.setOutputPath(job,new Path(args[1]));

            //提交作业到yarn中运行
            boolean b = job.waitForCompletion(true);

            if(b){
                System.out.println("========33期好友推荐案例1执行成功！！！！========");
                System.out.println("输出结果目标在hdfs路径："+args[1]);
            }else {
                System.out.println("-------------- 执行失败！！ -----------------");
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
